package com.zr.disruptor.impl;

import com.alibaba.fastjson.JSON;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.zr.domain.bar.entity.MarketDataEvent;
import com.zr.domain.bar.event.MarketDataEventFactory;
import com.zr.disruptor.handler.InnerDBHandler;
import com.zr.disruptor.handler.SendMessageHandler;
import com.zr.disruptor.handler.TransHandler;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.util.concurrent.ThreadFactory;

@Service
@Slf4j
public class DisruptorQueue implements Serializable, DisposableBean, InitializingBean, ApplicationContextAware {

    private static final int RING_BUFFER_SIZE = 1024 * 1024;

    @Autowired
    private transient InnerDBHandler innerDBHandler; // 标记为 transient（假设这些 Handler 不可序列化）

    @Autowired
    private transient SendMessageHandler sendMessageHandler;

    @Autowired
    private transient TransHandler transHandler;

    @Getter
    private transient Disruptor<MarketDataEvent> disruptor; // 标记为 transient

    @Getter
    private transient RingBuffer<MarketDataEvent> ringBuffer;

    private transient ApplicationContext applicationContext; // 用于反序列化后重新注入依赖

    @Override
    public void destroy() throws Exception {
        if (disruptor != null) {
            disruptor.shutdown();
            log.info("Disruptor of MarketDataEvent shutdown");
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        initializeDisruptor();
    }

    // 提取初始化逻辑到独立方法
    private void initializeDisruptor() {
        ThreadFactory threadFactory = r -> new Thread(r, "DisruptorThread");
        disruptor = new Disruptor<>(
                new MarketDataEventFactory(),
                RING_BUFFER_SIZE,
                threadFactory,
                ProducerType.SINGLE,
                new YieldingWaitStrategy()
        );

        // 重新注入依赖（反序列化后可能为 null）
        if (innerDBHandler == null) {
            innerDBHandler = applicationContext.getBean(InnerDBHandler.class);
        }
        if (sendMessageHandler == null) {
            sendMessageHandler = applicationContext.getBean(SendMessageHandler.class);
        }
        if (transHandler == null) {
            transHandler = applicationContext.getBean(TransHandler.class);
        }

        disruptor.handleEventsWith(innerDBHandler, transHandler)
                .then(sendMessageHandler);

        disruptor.start();
        ringBuffer = disruptor.getRingBuffer();
        log.info("Disruptor of MarketDataEvent initialized");
    }

    // 自定义序列化（仅序列化非 transient 字段）
    private void writeObject(ObjectOutputStream oos) throws IOException {
        oos.defaultWriteObject();
    }

    // 自定义反序列化后初始化
    private void readObject(ObjectInputStream ois) throws IOException, ClassNotFoundException {
        ois.defaultReadObject();
        // 手动触发依赖注入和初始化
        if (applicationContext != null) {
            applicationContext.getAutowireCapableBeanFactory().autowireBean(this);
            initializeDisruptor();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public void offer(MarketDataEvent t) {
        log.info("Publish MarketDataEvent [{}]",t);
        long sequence = ringBuffer.next();
        try {
            MarketDataEvent event = ringBuffer.get(sequence);
            BeanUtils.copyProperties(t, event);
        } finally {
            ringBuffer.publish(sequence);
        }
    }

    public MarketDataEvent pop() {
        long sequence = ringBuffer.next();
        try {
            return ringBuffer.get(sequence);
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}